package com.desmart.cdc.mysql;


import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class MysqlCDC {


    public static void main(String[] args) throws Exception {
        SourceFunction<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("10.0.12.201")
                .port(3306)
                .databaseList("cdc_db") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("cdc_db.user") // set captured table
                .username("cdcuser")
                .password("passw0rd")
                .deserializer(new StringDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .startupOptions(StartupOptions.initial())  // 从哪里开始
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 如果启用默认的 check point 表示依赖 Flink 的 check point 机制
        //env.enableCheckpointing(3000);



        env
                .addSource(mySqlSource)
                // set 4 parallel source tasks
                .print()
                .setParallelism(1); // use parallelism 1 for sink to keep message ordering

        env.execute("Print MySQL Snapshot + Binlog");
    }

}
